package com.ebspp.client.buffer;

import java.util.List;

import org.apache.thrift.TException;
import com.ebspp.client.placement.IPlacement;
import com.ebspp.client.serverManager.ServerManager;
import com.ebspp.global.EBSPPTimer;
import com.ebspp.thrift.DataMsg;
import com.ebspp.thrift.ResourceType;

/**
 * Writer is responsible for performing write operations to one target server
 * it analyzes the queue for the server and decides how much to write next time

 * @author x23pan
 *
 */
public class MvAsyncBufWriter implements Runnable
{
	boolean alive = true;
	
	MvAsyncBuffer buf;
	int bufID;
	String server = null;
	ResourceType resType;
	
	IPlacement placementCB = null;
	
	public MvAsyncBufWriter(MvAsyncBuffer buffer, String server)
	{
		this.buf = buffer;
		bufID = buf.getBufID();
		this.server = server;
		resType = buf.getResType();
		
		placementCB = buf.getPlacementCB();
	}
	
	
	@Override
	public void run() 
	{
		while(true)
		{
			try 
			{
				if(buf.getWQueueSize(server) > 0)
				{
					//decide how many data to write this time
					int writeNum = calWAmount();
					//decide if it needs more resource based on buffer pressure
					int proNum = calProvision();
					boolean proSucc = false;
					if(proNum != 0)
					{
						proSucc = ServerManager.wProvision(server, bufID, proNum);
						//if not success, here it might need to interrupt the bufferMonitor thread.
						if(ServerManager.getWConnRes(server, bufID) == 0)
						{
							close("Cannot write to server: " + server);
							return;
						}		
					}
					
					List<DataMsg> data = buf.readWQueue(server, writeNum);
					try 
					{
						List<DataMsg> failList = ServerManager.write(data, server, bufID);
						if(failList.size() > 0)
						{
							System.err.println("Write fail detected:" +failList.size());
							data.removeAll(failList);
						}				
						placementCB.setPlace(server, data);
						buf.cleanBufMsg(data);
					}
					catch (TException e) {
						// TODO Auto-generated catch block
						close(e.getMessage() + "\nUnable to write to the server");
						e.printStackTrace();
						return;
					}
				}			
				
				//here should be modified to a wait()/notify() based strategy
				Thread.sleep(EBSPPTimer.asyncBuffWriteT);
			} 
			catch (InterruptedException e) {
				// TODO Auto-generated catch block
				close("Interruption for buffer pool write wait");
				e.printStackTrace();
				return;
			}
		}
	}
	
	/**
	 * calculate how many data to write the server
	 * @return
	 */
	private int calWAmount()
	{
		//naive solution, write all
		return Integer.MAX_VALUE;
	}
	
	/**
	 * calculate how much more resource to provision
	 * @return
	 */
	private int calProvision()
	{
		int used = ServerManager.getWConnRes(server, bufID);
		//naive solution. If none, provision 10, otherwise, provision 0
		if(used == 0)
			return 10;
		else
			return 0;
	}
	
	public void close(String msg)
	{
		System.err.println(msg);
		alive = false;
	}
}
